package org.shj.spark.operator;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;

public class FlatMapOperator {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("FlatMapOperator");
		JavaSparkContext ctx = new JavaSparkContext(conf);
		
		List<String> lineList = Arrays.asList("hello zhuyin", "hello zhuyin", "hello liangyongqi");
		JavaRDD<String> linesRDD = ctx.parallelize(lineList);
		
		//flatMap = flat + map
		JavaRDD<String> words = linesRDD.flatMap(new FlatMapFunction<String, String>(){
			private static final long serialVersionUID = 8619005120758204134L;

			public Iterator<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" ")).iterator();
			}
			
		});
		
		words.foreach(new VoidFunction<String>(){
			private static final long serialVersionUID = 462595944677020583L;

			@Override
			public void call(String result) throws Exception {
				System.out.println(result);
				
			}
			
		});
		
		ctx.close();
		
	}

}
